package com.shujia.mr.kqzl2;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;

public class KqzlMapper2 extends Mapper<LongWritable, Text, Text, Text> {

    public HashMap<String, String> map = new HashMap<>();

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        //在整个map任务执行之前执行
        //适合读取一些小文件的数据，在map任务运行过程中进行关联
        //若文件较大，例如10G以上，每个map任务都会读取一次这个文件，就会导致map任务的运行时间很长
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        //进入到map方法中的一行数据可能来自于上一次的结果数据或者地点city数据
        //        20180101-1279   128
        //        1279,高新区,合肥
        //我们需要先判断进入到map方法的数据来自于哪里？
        //这个哪里，实际上map任务处理的数据来自于切片，切片实际上来自于block块，而block块是属于文件的一部分
        //可以获取到数据所处的最终文件的名字
        FileSplit fileSplit = (FileSplit) context.getInputSplit(); //获取当前数据所在的切片，向下转型
        String fileName = fileSplit.getPath().getName(); //获取切片所在的路径的名称
        //文件的名字：city开头，part开头
        if(fileName.startsWith("city")){
            //说明这条数据来自于城市表
            String[] datas = value.toString().split(",");
            context.write(new Text(datas[0]),new Text("#"+datas[2]+"-"+datas[1]));
        }else if(fileName.startsWith("part")){
            String[] datas = value.toString().split("\t");
            String dateTime = datas[0].split("-")[0];
            String id = datas[0].split("-")[1];
            String pm25Avg = datas[1];

            context.write(new Text(id),new Text("$"+dateTime+"-"+pm25Avg));
        }

    }
}
